🟡 WorkflowServer status can report "running" even after handler completion if no terminal event was observed

In _WorkflowHandler.status, if no terminal event was recorded and run_handler.is_done() is true, the code returns "running" unconditionally.

Actual behavior: completed/failed/cancelled runs can be reported as "running" in persistence/API if the terminal StopEvent was not observed/recorded by _stream_events for any reason.

Expected behavior: if the handler is done and there is no terminal event, the status should be derived from handler completion state (cancelled vs exception vs completed), not forced to "running".

Code: workflows/server/server.py:1685-1713

Recommendation: When run_handler.is_done() is true and _terminal_event is None, fall back to run_handler.cancelled() / run_handler.exception() to classify as cancelled/failed/completed (or at least "failed"/"cancelled"), rather than returning "running".

Was this helpful? React with 👍 or 👎 to provide feedback.

🔴 BasicRuntime concurrency limiting can silently stop working due to WeakValueDictionary semaphore storage

BasicRuntime stores per-workflow semaphores in a weakref.WeakValueDictionary:

self._max_concurrent_runs: weakref.WeakValueDictionary[str, asyncio.Semaphore]

(basic.py:184-188).

Because the only long-lived reference to each semaphore is the weak dictionary entry, semaphores may be garbage-collected at any time when not currently being awaited. When that happens, the next call to _maybe_acquire_max_concurrent_runs() will create a new semaphore, effectively resetting concurrency limits and allowing more than the configured num_concurrent_runs.

Actual: concurrency limit can be bypassed intermittently/non-deterministically.
Expected: concurrency limit should be enforced consistently for the process lifetime (or at least until runtime.destroy()).

Impact: can exceed intended concurrency caps, causing resource exhaustion and incorrect load-shedding behavior.

Recommendation: Use a normal dict (strong refs) for _max_concurrent_runs, and clear it in destroy(); or otherwise keep strong references for semaphore lifetime management.

Was this helpful? React with 👍 or 👎 to provide feedback.

🟡 Idle release uses graceful cancellation even when graceful=False, contradicting intended semantics

_WorkflowHandler.cancel_handlers_and_tasks(graceful=False) is documented/used for idle release “where we don't want to emit cancel event”, but it still calls await self.run_handler.cancel_run() before hard-cancelling.

Actual: idle release sends a TickCancelRun into the workflow (cancel_run()), which can cause the workflow to emit WorkflowCancelledEvent and transition persisted status to cancelled.
Expected: idle release should stop in-memory execution without changing logical workflow outcome (it should remain resumable/running), or at least avoid emitting cancellation signals.

Code:

• In non-graceful branch:

await self.run_handler.cancel_run()
...
self.run_handler.cancel()

workflows/server/server.py:1937-1945

Impact: idle release may incorrectly cancel runs instead of just unloading them, breaking resumability and causing incorrect persisted status.

Recommendation: For graceful=False, avoid calling cancel_run(); instead only stop the streaming task and close adapter/resources (or implement a runtime-specific ‘detach/unload’ that doesn’t enqueue TickCancelRun).

Was this helpful? React with 👍 or 👎 to provide feedback.

🔴 Database connection leak when exception raised in edit_state context

When an exception is raised inside the edit_state context manager, the database connection is never closed, causing a resource leak.

async with store.edit_state() as state:
    state["value"] = "modified"
    raise ValueError("intentional error")  # Connection leaks!

Recommendation: Add rollback and connection cleanup in the exception handler:

async with self._lock:
    state, commit_fn = await self._run_sync(_edit_with_lock)
    try:
        yield state
        await self._run_sync(commit_fn, state)
    except Exception:
        # Need to rollback and close the connection
        def _rollback_and_close() -> None:
            try:
                trans.rollback()
            finally:
                conn.close()
        await self._run_sync(_rollback_and_close)
        raise

Alternatively, restructure to use a separate cleanup function that's always called, or capture conn and trans in a way that allows cleanup in the except block.

Was this helpful? React with 👍 or 👎 to provide feedback.